package com.mengshuai.technology.order.rocketmq.producer;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

/**
 * rocketmq 生产者服务
 *
 * @author ms
 * @since 2025/06/25
 */
@Service
public class RocketMQProducerService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步发送消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param key   键
     * @param msg   消息
     * @return 发送结果
     */
    public SendResult sendSyncMessage(String topic, String tag, String key, Object msg) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_KEYS, key)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .build();
        SendResult sendResult = rocketMQTemplate.syncSend(topic + ":" + tag, message);
        System.out.println("发送结果：" + sendResult);
        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            System.out.println("发送成功：" + sendResult);
        } else {
            System.out.println("发送失败：" + sendResult);
        }
        return sendResult;
    }

    /**
     * 异步发送消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param key   键
     * @param msg   消息
     */
    public void sendAsyncMessage(String topic, String tag, String key, Object msg) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_KEYS, key)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .build();
        rocketMQTemplate.asyncSend(topic + ":" + tag, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                try {
                    Thread.sleep(1000L * 5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("发送成功");
            }

            @Override
            public void onException(Throwable e) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
                System.out.println("发送失败");
            }
        });
        System.out.println("the method sendAsyncMessage end");
    }

    /**
     * 发送事务消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param key   键
     * @param msg   消息
     */
    public void sendTransactionMessage(String topic, String tag, String key, Object msg) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_KEYS, key)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .build();
        rocketMQTemplate.sendMessageInTransaction(topic + ":" + tag, message, null);
    }
}
